package net.bwie.realtime.jtp.dws.log.function;

import net.bwie.realtime.jtp.dws.log.bean.PageViewBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {

    // 格式实现
    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
    @Override
    public void apply(String s,
                      TimeWindow timeWindow,
                      Iterable<PageViewBean> iterable,
                      Collector<String> collector) throws Exception {

        // 开始时间 和 结束时间
        String windowStartTime = this.format.format(timeWindow.getStart());
        String windowEndTime = this.format.format(timeWindow.getEnd());

        Long pvCount = 0L;
        Long pvDuringTime = 0L;
        Long uvCount = 0L;
        Long sessionCount = 0L;
        for (PageViewBean pageViewBean : iterable) {
            pvCount += pageViewBean.getPvCount();
            pvDuringTime += pageViewBean.getPvDuringTime();
            uvCount += pageViewBean.getUvCount();
            sessionCount += pageViewBean.getSessionCount();
        }
        //输出
        String output=
                windowStartTime+","+
                windowEndTime+","+
                windowStartTime.substring(0,10)+ "," +
                        //brand, channel, province, is_new 字段
                s+","+
                uvCount+","+
                sessionCount+","+
                pvCount
                +","+pvDuringTime;
        collector.collect(output);
    }
}











